%% copy riak-0.14.2 source to learn dynamo implementation and erlang
%% author: lichuang
%% copy source from vclock.erl

%% @doc A simple Erlang implementation of vector clocks as inspired by Lamport logical clocks.
%%
%% @reference Leslie Lamport (1978). "Time, clocks, and the ordering of events
%% in a distributed system". Communications of the ACM 21 (7): 558-565.
%%
%% @reference Friedemann Mattern (1988). "Virtual Time and Global States of
%% Distributed Systems". Workshop on Parallel and Distributed Algorithms:
%% pp. 215-226

-module(vclock).

-export([fresh/0, descends/2, merge/1, get_counter/2, get_timestamp/2,
	       increment/2, increment/3, all_nodes/1, equal/2, prune/3, timestamp/0]).

-type vclock() :: [vc_entry()].
% The timestamp is present but not used, in case a client wishes to inspect it
-type vc_entry() :: {vclock_node(), {counter(), timestamp()}}.

% Nodes can have any term() as a name, but they must differ from each other.
-type vclock_node() :: term().
-type counter() :: integer().
-type timestamp() :: integer().

-spec fresh() -> vclock().
fresh() ->
  [].

% @doc Increment VClock at Node.
-spec increment(Node :: vclock_node(), VClock :: vclock()) -> vclock().
increment(Node, VClock) ->
  increment(Node, timestamp(), VClock).

% @doc Increment VClock at Node.
-spec increment(Node :: vclock_node(), IncTs :: timestamp(),
                VClock :: vclock()) -> vclock().
increment(Node, IncTs, VClock) ->
  {{_Ctr, _TS} = C1, NewV} = case lists:keytake(Node, 1, VClock) of
    false ->
      {{1, IncTs}, VClock};
    {value, {_N, {C, _T}}, ModV} ->
      {{C + 1, IncTs}, ModV}
  end,
  [{Node, C1} | NewV].

% @doc Return a timestamp for a vector clock
-spec timestamp() -> timestamp().
timestamp() ->
  calendar:datetime_to_gregorian_seconds(erlang:universaltime()).

% @doc Combine all VClocks in the input list into their least possible
%      common descendant.
-spec merge(VClocks :: [vclock()]) -> vclock().
merge([]) -> [];
merge([SingleVclock]) -> SingleVclock;
merge([First|Rest])   -> merge(Rest, lists:keysort(1, First)).

merge([], NClock) -> NClock;
merge([AClock|VClocks], NClock) ->
  merge(VClocks, merge(lists:keysort(1, AClock), NClock, [])).

merge([], [], AccClock) -> lists:reverse(AccClock);
merge([], [Left|Rest], AccClock) -> merge([], Rest, [Left | AccClock]);
merge(Left, [], AccClock) -> merge([], Left, AccClock);
merge(V=[{Node1, {Ctr1, TS1}} | VClock],
      N=[{Node2, {Ctr2, TS2}} | NClock], AccClock) ->
  if Node1 < Node2 ->
      merge(VClock, N, [{Node1, {Ctr1, TS1}} | AccClock]);
     Node1 > Node2 ->
       merge(V, NClock, [{Node2, {Ctr2, TS2}} | AccClock]);
     true ->
       ({_Ctr, _TS} = C1) = if Ctr1 > Ctr2 -> {Ctr1, TS1};
                              true -> {Ctr2, TS2}
                            end,
       merge(VClock, NClock, [{Node1, C1} | AccClock])
  end.       

% @doc Return true if Va is a direct descendant of Vb, else false -- remember, a vclock is its own descendant!
-spec descends(Va :: vclock(), Vb :: vclock()) -> boolean().
descends(_, []) ->
  % all vclocks descend from the empty vclock
  true;
descends(Va, Vb) ->
  [{NodeB, {CtrB, _T}}|RestB] = Vb,
  CtrA = 
  case proplists:get_value(NodeB, Va) of
    undefined ->
      false;
    {CA, _TSA} -> CA
  end,
  case CtrA of
    false -> false;
    _ -> 
      if
        CtrA < CtrB ->
          false;
        true ->
          descends(Va,RestB)
      end
  end.

% @doc Return the list of all nodes that have ever incremented VClock.
-spec all_nodes(VClock :: vclock()) -> [vclock_node()].
all_nodes(VClock) ->
  [X || {X,{_,_}} <- VClock].

% @doc Compares two VClocks for equality.
%      Not very fast.
-spec equal(VClockA :: vclock(), VClockB :: vclock()) -> boolean().
equal(VA, VB) ->
  VSet1 = sets:from_list(VA),
  VSet2 = sets:from_list(VB),
  case sets:size(sets:subtract(VSet1, VSet2)) > 0 of
    true -> false;
    false ->
      case sets:size(sets:subtract(VSet2, VSet1)) > 0 of
        true -> false;
        false -> true
      end
  end.      

% @doc Get the counter value in VClock set from Node.
-spec get_counter(Node :: vclock_node(), VClock :: vclock()) -> counter() | undefined.
get_counter(Node, VClock) ->
  case proplists:get_value(Node, VClock) of
    {Ctr, _TS} -> Ctr;
    undefined  -> undefined
  end.      

% @doc Get the timestamp value in a VClock set from Node.
-spec get_timestamp(Node :: vclock_node(), VClock :: vclock()) -> timestamp() | undefined.
get_timestamp(Node, VClock) ->
  case proplists:get_value(Node, VClock) of
    {_Ctr, TS} -> TS;
    undefined -> undefined
  end.

% @doc Possibly shrink the size of a vclock, depending on current age and size.
-spec prune(V::vclock(), Now::integer(), BucketProps::term()) -> vclock().
prune(V, Now, BucketProps) ->
  SortV = lists:sort(fun({_, {_, A}}, {_, {_, B}}) -> A < B end, V),
  prune_vclock1(SortV, Now, BucketProps).

% @private
prune_vclock1(V, Now, BProps) ->
  case length(V) =< proplists:get_value(small_vclock, BProps) of
    true -> V;
    false ->
      {_, {_, HeadTime}} = hd(V),
      case (Now - HeadTime) < proplists:get_value(young_vclock, BProps) of
        true -> V;
        false -> prune_vclock1(V, Now, BProps, HeadTime)
      end
  end.      
% @private
prune_vclock1(V,Now,BProps,HeadTime) ->
  % has a precondition that V is longer than small and older than young
  case length(V) > proplists:get_value(big_vclock, BProps) of
    true ->
      prune_vclock1(tl(V), Now, BProps);
    false ->
      case (Now - HeadTime) > proplists:get_value(old_vclock, BProps) of
        true ->
          prune_vclock1(tl(V), Now, BProps);
        false ->
          V
      end
  end.      
